#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/time.h>
#include <thread>
#include <map>
#include <mutex>
#include <sys/prctl.h>
#include "memfd_share.h"

#define SHAREDATA_SUB_BUF_SIZE 1024 * 1024 * 100 //subscriber初始化需要预分配的空间
#define SHAREDATA_SUB_PACK_SIZE 1024 * 1024 * 2  //subscriber初始化预留的单笔数据大小

#define SHAREDATA_PUB_BUF_SIZE 1024 * 1024 * 100 //publish初始化需要预分配的空间
#define SHAREDATA_PUB_PACK_SIZE 1024 * 1024 * 2 //publish共享内存后面预留的单笔数据大小

namespace share
{
      typedef struct tagTShareData
      {
            char topic[256];
            int threadExit;
            int subInitFlag;
            int pubInitFlag;
            int shareDataSize;
            int singlePackSize;
            ShareCallbackFun pFuncCallBack;
            std::thread *subDataThread;
            void *pUserData;
            CProcessShare *pPubShareFileHandle;
            CProcessShare *pSubShareFileHandle;
      } TShareData;

      std::map<const char *, TShareData *> g_mapTShareData;
      std::mutex g_mapLock;
      int g_dwRefCnt = 0; /*TShareData 引用计数，最后一个执行销毁动作*/

      int MEMFD_Init()
      {
            return 0;
      }

      static void handleSubscribeData(TShareData *pShareDataCtx)
      {
            unsigned char *buf = NULL;
            unsigned int len = 0;
            bool bRet = false;
            char topicPath[512] = {0};

            /*设置线程名字*/
            prctl(PR_SET_NAME, __func__);

            if (pShareDataCtx == NULL)
            {
                  printf(" thread exit! \n");
                  return;
            }

            snprintf(topicPath, sizeof(topicPath), "%s_%s", TZTEK_PROCESS_SHAREFILE_PATH, pShareDataCtx->topic);
            while (pShareDataCtx->threadExit == false)
            {
                  struct timeval t1, t2;
                  gettimeofday(&t1, NULL);
                  if (pShareDataCtx->subInitFlag == 0)
                  {
                        bRet = pShareDataCtx->pSubShareFileHandle->Init(topicPath, pShareDataCtx->shareDataSize, pShareDataCtx->topic, pShareDataCtx->singlePackSize, false);
                        if (bRet == false)
                        {
                              printf(" thread exit!  topic:%s  subscribe init failed!\n", pShareDataCtx->topic);
                              return;
                        }
                        printf(" topic:%s  sub sharefile init. pShareDataCtx->pSubShareFileHandle=%p\n", topicPath, pShareDataCtx->pSubShareFileHandle);
                        pShareDataCtx->subInitFlag = 1; /*设置订阅初始化完成标志*/
                  }
                  if ((pShareDataCtx->pSubShareFileHandle)->GetData(buf, len))
                  {
                        printf(" topic:%s , got data Len:%d\n", pShareDataCtx->topic, len);
                        pShareDataCtx->pSubShareFileHandle->UpdateReadIdx(len);
                        if (0 != len)
                        {
                              gettimeofday(&t2, NULL);
                              printf(" sub topic:%s  cost:%d ms. \n", pShareDataCtx->topic, (int)((t2.tv_sec - t1.tv_sec) * 1000 + (t2.tv_usec - t1.tv_usec) / 1000));
                              pShareDataCtx->pFuncCallBack(buf, len, pShareDataCtx->pUserData);

                              gettimeofday(&t1, NULL);
                              printf(" sub topic:%s  callback cost:%d ms. \n", pShareDataCtx->topic, (int)((t1.tv_sec - t2.tv_sec) * 1000 + (t1.tv_usec - t2.tv_usec) / 1000));
                        }
                        //根据出图的速度，收的太快耗资源
                        usleep(20*1000);
                  } else
                  {
                        usleep(20*1000);
                  }
            }
            printf("topic:%s thread exit! \n", pShareDataCtx->topic);
            return;
      }

      static bool memfd_data_get(const char *pTopic, TShareData **pCtx)
      {
            bool bRet = false;
            struct timeval t1, t2;
            gettimeofday(&t1, NULL);

            if (g_mapTShareData.size() == 0)
            {
                  return false;
            }
            g_mapLock.lock();
            auto itor = g_mapTShareData.find(pTopic);
            if (itor == g_mapTShareData.end())
            {
                  printf(" not found topic:%s ,now  create \n", pTopic);
                  bRet = false;
            }
            else
            {
                  *pCtx = itor->second;
                  printf(" found topic:%s \n", itor->second->topic);
                  bRet = true;
            }
            g_mapLock.unlock();
            gettimeofday(&t2, NULL);
            printf("  topic:%s searching cost:%d ms. \n", pTopic, (int)((t2.tv_sec - t1.tv_sec) * 1000 + (t2.tv_usec - t1.tv_usec) / 1000));
            return bRet;
      }

      static bool memfd_sub_init(const char *pTopic, TShareData **pCtxSh)
      {
            TShareData *pCtx = NULL;

            if (pTopic == NULL)
            {
                  printf(" invalid parameters!\n");
                  return false;
            }

            if (memfd_data_get(pTopic, &pCtx) == false)
            {
                  *pCtxSh = (TShareData *)malloc(sizeof(TShareData));
                  if (*pCtxSh == NULL)
                  {
                        printf(" topic:%s new ctx failed!\n", pTopic);
                        return false;
                  }
                  pCtx = *pCtxSh;
                  pCtx->pSubShareFileHandle = new (std::nothrow) CProcessShare();
                  if (pCtx->pSubShareFileHandle == NULL)
                  {
                        printf(" new CProcessShare failed.\n");
                        return false;
                  }

                  pCtx->subInitFlag = 0; /*尚未初始化完成. 只是申请了一个sharedata 句柄*/
                  pCtx->pubInitFlag = 0; /*尚未初始化完成. 只是申请了一个sharedata 句柄*/

                  memset(pCtx->topic, 0, sizeof(pCtx->topic));
                  strncpy(pCtx->topic, pTopic, sizeof(pCtx->topic) - 1);
                  pCtx->pPubShareFileHandle = NULL;
                  pCtx->threadExit = false;
                  pCtx->pUserData = NULL;
                  pCtx->subDataThread = nullptr;
                  pCtx->shareDataSize = SHAREDATA_SUB_BUF_SIZE;
                  pCtx->singlePackSize = SHAREDATA_SUB_PACK_SIZE;
                  g_mapLock.lock();
                  g_mapTShareData[pTopic] = pCtx;
                  g_mapLock.unlock();
                  /*更新资源引用计数。 当g_dwRefCnt == 0 才释放所有资源*/
                  g_dwRefCnt++;

                  printf(" [topic]:%s Init successfully.\n", pTopic);
            }
            else
            {
                  *pCtxSh = pCtx;
                  printf(" [topic]:%s got ctx.\n", pTopic);
            }

            return true;
      }

      static bool memfd_pub_init(const char *pTopic, TShareData **pCtxSh, int setFlag)
      {
            bool bRet = false;
            char topicPath[256] = {0};
            TShareData *pCtx = NULL;

            if (pTopic == NULL)
            {
                  printf(" invalid parameters!\n");
                  return false;
            }
            snprintf(topicPath, sizeof(topicPath), "%s_%s", TZTEK_PROCESS_SHAREFILE_PATH, pTopic);

            if (memfd_data_get(pTopic, &pCtx) == false)
            {
                  *pCtxSh = (TShareData *)malloc(sizeof(TShareData));
                  if (*pCtxSh == NULL)
                  {
                        printf(" topic:%s new ctx failed!\n", pTopic);
                        return false;
                  }
                  pCtx = *pCtxSh;
                  pCtx->pPubShareFileHandle = new (std::nothrow) CProcessShare();
                  if (pCtx->pPubShareFileHandle == NULL)
                  {
                        printf(" new CProcessShare failed.\n");
                        return false;
                  }
                  pCtx->pubInitFlag = 0; /*尚未初始化完成. 只是申请了一个sharedata 句柄*/
                  pCtx->subInitFlag = 0; /*尚未初始化完成. 只是申请了一个sharedata 句柄*/
                  if (setFlag == 0)
                  {
                        bRet = pCtx->pPubShareFileHandle->Init(topicPath, SHAREDATA_SUB_BUF_SIZE, pTopic, SHAREDATA_SUB_PACK_SIZE, true);
                        if (bRet == false)
                        {
                              printf(" topic:%s init failed.\n", pTopic);
                              return false;
                        }
                        printf(" topic:%s pub sharefile init. pCtx=%p, pPubShareFileHandle=%p!\n", pTopic, pCtx, pCtx->pPubShareFileHandle);
                        pCtx->pubInitFlag = 1;
                  }
                  memset(pCtx->topic, 0, sizeof(pCtx->topic));
                  strncpy(pCtx->topic, pTopic, sizeof(pCtx->topic) - 1);
                  pCtx->pSubShareFileHandle = NULL;
                  pCtx->threadExit = false;
                  pCtx->pUserData = NULL;
                  pCtx->subDataThread = nullptr;
                  pCtx->shareDataSize = SHAREDATA_SUB_BUF_SIZE;
                  pCtx->singlePackSize = SHAREDATA_SUB_PACK_SIZE;

                  g_mapLock.lock();
                  g_mapTShareData[pTopic] = pCtx;
                  g_mapLock.unlock();
                  /*更新资源引用计数。 当g_dwRefCnt == 0 才释放所有资源*/
                  if (setFlag == 0)
                  {
                        g_dwRefCnt++;
                  }
                  printf(" topic:%s Init successfully.\n", pTopic);
            }
            else
            {
                  if (pCtx->pubInitFlag == 0)
                  {
                        pCtx->pPubShareFileHandle = new (std::nothrow) CProcessShare();
                        if (pCtx->pPubShareFileHandle == NULL)
                        {
                              printf(" new CProcessShare failed.\n");
                              return false;
                        }
                        printf(" topic:%s pub sharefile later init. pCtx=%p, pPubShareFileHandle=%p!\n", pCtx->topic, pCtx, pCtx->pPubShareFileHandle);
                        printf(" topic:%s, shareDataSize=%d, singlePackSize=%d.\n", pCtx->topic, pCtx->shareDataSize, pCtx->singlePackSize);
                        bRet = pCtx->pPubShareFileHandle->Init(topicPath, pCtx->shareDataSize, topicPath, pCtx->singlePackSize, true);
                        if (bRet == false)
                        {
                              printf(" topic:%s init failed.\n", pTopic);
                              return false;
                        }
                        pCtx->pubInitFlag = 1; /*初始化完成*/
                        /*更新资源引用计数。 当g_dwRefCnt == 0 才释放所有资源*/
                        g_dwRefCnt++;
                  }
                  *pCtxSh = pCtx;
                  printf(" topic:%s got ctx=%p.\n", pTopic, pCtxSh);
            }

            return true;
      }

      int MEMFD_Subscriber(const char *pTopic, ShareCallbackFun callBack, void *pUserData)
      {
            TShareData *pCtx = NULL;
            if (pTopic == NULL)
            {
                  printf("pTopic  is null. \n");
                  return -1;
            }
            if (callBack == NULL)
            {
                  printf(" topic:%s, pFuncCallBack is null. \n", pTopic);
                  return -1;
            }
            struct timeval t1, t2;
            gettimeofday(&t1, NULL);
            if (memfd_sub_init(pTopic, &pCtx) == false)
            {
                  printf("  Subscribe topic:%s failed,  need call Init firstly. \n", pTopic);
                  return -1;
            }
            gettimeofday(&t2, NULL);
            printf("  Sub topic:%s init cost:%d ms. \n", pTopic, (int)((t2.tv_sec - t1.tv_sec) * 1000 + (t2.tv_usec - t1.tv_usec) / 1000));
            pCtx->pFuncCallBack = callBack;
            pCtx->pUserData = pUserData;
            if (pCtx->pSubShareFileHandle == NULL)
            {
                  printf("  topic:%s, pShareFileHandle is null.\n", pTopic);
                  return -1;
            }
            pCtx->subDataThread = new std::thread(handleSubscribeData, pCtx);
            pCtx->subDataThread->detach();
            printf(" ****Subscribe topic:%s successfully ****\n\n", pTopic);
            return 0;
      }

      int MEMFD_Publisher(const char *pTopic, unsigned char *pData, unsigned int dwLen)
      {
            bool bRet = false;
            TShareData *pCtx = NULL;

            if (pTopic == NULL || pData == NULL || dwLen == 0)
            {
                  printf(" topic : %s invalid parameters.\n",pTopic);
                  return -1;
            }
            struct timeval t1, t2;
            gettimeofday(&t1, NULL);
            if (memfd_pub_init(pTopic, &pCtx, 0) == false)
            {
                  printf(" topic:%s, get  ctx failed. Maybe need call Init function firstly.\n", pTopic);
                  return -1;
            }

            if (pCtx == NULL || pCtx->pPubShareFileHandle == NULL)
            {
                  printf(" topic:%s, pCtx or pShareFileHandle is null. Maybe need call Init function firstly.\n", pTopic);
                  return -1;
            }

            bRet = pCtx->pPubShareFileHandle->InputData(pData, dwLen);
            if (bRet == false)
            {
                  printf(" topic:%s publish data failed.\n", pTopic);
                  return -1;
            }

            gettimeofday(&t2, NULL);
            printf("  pub topic:%s  cost:%d ms. \n", pTopic, (int)((t2.tv_sec - t1.tv_sec) * 1000 + (t2.tv_usec - t1.tv_usec) / 1000));
            return 0;
      }

      int MEMFD_Publisher2(const char *pTopic, unsigned char *pData, unsigned int dwLen, unsigned char *pData2, unsigned int dwLen2)
      {
            bool bRet = false;
            TShareData *pCtx = NULL;

            if (pTopic == NULL || pData == NULL || dwLen == 0)
            {
                  printf("  invalid parameters topic :%s\n",pTopic);
                  return -1;
            }

            if (memfd_pub_init(pTopic, &pCtx, 0) == false)
            {
                  printf("  topic:%s, get  ctx failed. Maybe need call Init function firstly.\n", pTopic);
                  return -1;
            }

            if (pCtx->pPubShareFileHandle == NULL)
            {
                  printf("  topic:%s, pShareFileHandle is null. Maybe need call Init function firstly.\n", pTopic);
                  return -1;
            }

            bRet = pCtx->pPubShareFileHandle->InputData(pData, dwLen, pData2, dwLen2);
            if (bRet == false)
            {
                  printf("  topic:%s publish data failed.\n", pTopic);
                  return -1;
            }

            return 0;
      }

      int MEMFD_Adv_Parameter_Set(const char *pTopic, unsigned int dwMaxBufSize, unsigned int dwSinglePackSize)
      {
            int iRet = -1;
            TShareData *pCtx = NULL;

            if (pTopic == NULL || dwMaxBufSize == 0 || dwSinglePackSize == 0)
            {
                  printf("  topic:%s invalid paramters!\n", pTopic);
                  return -1;
            }

            /*先查找ctx是否已经存在，不存在则预初始化*/
            if (memfd_data_get(pTopic, &pCtx) == false)
            {
                  /*只初始化ctx, 不初始化共享内存，方便设定高级参数*/
                  if (memfd_pub_init(pTopic, &pCtx, 1) == false)
                  {
                        printf("  topic:%s, get  ctx failed. Maybe need call Init function firstly.\n", pTopic);
                        return -1;
                  }
            }
            /*修改高级参数*/
            pCtx->shareDataSize = dwMaxBufSize;
            pCtx->singlePackSize = dwSinglePackSize;
            printf("  topic:%s, fix to : shareDataSize=%d, singlePackSize=%d.\n", pCtx->topic, pCtx->shareDataSize, pCtx->singlePackSize);
            iRet = 0;

            return iRet;
      }

      int MEMFD_Release(void)
      {
            std::map<const char *, TShareData *>::iterator it2;

            if (g_dwRefCnt != 0)
            {
                  printf(" need wait all other topic exit.\n");
                  return 0;
            }

            auto itor = g_mapTShareData.begin();
            while (itor != g_mapTShareData.end())
            {
                  it2 = itor++;
                  if (it2->second->pPubShareFileHandle)
                  {
                        it2->second->pPubShareFileHandle->Release();
                        delete (it2->second->pPubShareFileHandle);
                  }
                  if (it2->second->pSubShareFileHandle)
                  {
                        it2->second->pSubShareFileHandle->Release();
                        delete (it2->second->pSubShareFileHandle);
                  }
                  free(it2->second);
                  g_mapTShareData.erase(it2);
            }

            printf(" release successfully.\n");
            return 0;
      }

}
